package org.study.canal.boot.starter;

import com.alibaba.fastjson2.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.*;
import org.springframework.stereotype.Component;
import org.study.canal.boot.config.CanalProperties;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

/**
 * 参考 https://blog.csdn.net/qq_43791139/article/details/107752629
 *
 * @author Administrator
 * @date 2021-12-23
 */
@Slf4j
@Component
public class CanalApplicationRunner implements ApplicationRunner {
    /**
     * 存储 sql 的队列
     */
    private Queue<String> SQL_QUEUE = new ConcurrentLinkedQueue<>();

    @Resource
    private CanalProperties canalProperties;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        String hostIp = AddressUtils.getHostIp();
        String hostName = AddressUtils.getHostName();
        InetAddress hostAddress = AddressUtils.getHostAddress();
        log.info("执行了自定义的 ApplicationRunner。hostIp={}, hostName={}, hostAddress={}", hostIp, hostName, JSON.toJSONString(hostAddress));

        InetSocketAddress address = new InetSocketAddress(canalProperties.getHostname(), canalProperties.getPort());
        CanalConnector connector = CanalConnectors.newSingleConnector(address, canalProperties.getDestination(), canalProperties.getUsername(), canalProperties.getPassword());
        try {
            // 打开连接
            connector.connect();
            // 指定要订阅的数据库表、表，格式为：数据库.表名,数据库.表名
            connector.subscribe(canalProperties.getFilter());
            // 回滚到未进行 ack 的地方，下次 fetch 时可以从最后一个没有 ack 的地方开始
            connector.rollback();
            while (true) {
                // 获取指定数量的数据：尝试从 binlog 中拉取 batchSize 条记录，有多少取多少
                Message message = connector.getWithoutAck(canalProperties.getBatchSize());
                // 获取批量ID
                long batchId = message.getId();
                // 获取批量的数量
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    // 如果没有数据，线程休眠 1s
                    TimeUnit.SECONDS.sleep(1L);
                } else {
                    // 如果有数据，处理数据
                    this.dataHandle(message.getEntries());
                }
                // 提交确认：对 batchId 进行确认，确认之后，小于等于此 batchId 的 Message 都会被确认
                connector.ack(batchId);
                // 处理失败，回滚数据，后续重新获取数据
                // connector.rollback(batchId);
                // 当队列中的 sql 达到一定数值时就执行
                if (SQL_QUEUE.size() >= 1) {
                    // 模拟执行队列里面的 sql 语句
                    int queueSize = SQL_QUEUE.size();
                    for (int i = 0; i < queueSize; i++) {
                        String sql = SQL_QUEUE.poll();
                        log.info("当前 sql 语句：{}", sql);
                        this.execute(sql);
                    }
                }
            }
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 数据处理
     *
     * @param entryList
     */
    private void dataHandle(List<CanalEntry.Entry> entryList) throws InvalidProtocolBufferException {
        for (CanalEntry.Entry entry : entryList) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                // 开启、关闭事务的实体类型：跳过
                continue;
            }
            // 获取发生变化的数据：RowChange 对象包含了一行数据变化的所有特征
            // 如 isDdl 表示是否为 ddl 操作，beforeColumns 表示变更前的数据字段，afterColumns 表示变更后的数据字段等
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

            // 获取事件类型：INSERT、UPDATE、DELETE 类型
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.debug("判断是否为 DDL 语句。isDdl={}，sql={}", rowChange.getIsDdl(), rowChange.getSql());

            CanalEntry.Header header = entry.getHeader();
            String formatMsg = String.format("logfileName=%s，logfileOffset=%s，schemaName=%s，tableName=%s，eventType=%s",
                    // 获取发生变化的 binlog 名称
                    header.getLogfileName(),
                    // 获取发生变化的 binlog 偏移量
                    header.getLogfileOffset(),
                    // 获取发生变化的数据库名称
                    header.getSchemaName(),
                    // 获取发生变化的表名称
                    header.getTableName(),
                    // 获取事件类型
                    eventType);
            log.info(formatMsg);

            if (eventType == CanalEntry.EventType.INSERT) {
                // 如果是 INSERT 语句
                this.saveInsertSql(entry);
            } else if (eventType == CanalEntry.EventType.DELETE) {
                // 如果是 DELETE 语句
                this.saveDeleteSql(entry);
            } else if (eventType == CanalEntry.EventType.UPDATE) {
                // 如果是 UPDATE 语句
                this.saveUpdateSql(entry);
            } else {
                // 其他语句
                log.warn("其他(除 INSERT、DELETE、UPDATE 外)类型的事件");
            }

            /*if (CanalEntry.EntryType.ROWDATA == entry.getEntryType()) {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                CanalEntry.EventType eventType = rowChange.getEventType();
                if (eventType == CanalEntry.EventType.INSERT) {
                    // 如果是 INSERT 语句
                    this.saveInsertSql(entry);
                } else if (eventType == CanalEntry.EventType.DELETE) {
                    // 如果是 DELETE 语句
                    this.saveDeleteSql(entry);
                } else if (eventType == CanalEntry.EventType.UPDATE) {
                    // 如果是 UPDATE 语句
                    this.saveUpdateSql(entry);
                } else {
                    // 其他语句
                    log.warn("其他(除 INSERT、DELETE、UPDATE 外)类型的事件");
                }
            }*/
        }
    }

    /**
     * 保存插入语句
     *
     * @param entry
     */
    private void saveInsertSql(CanalEntry.Entry entry) {
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDataList) {
                List<CanalEntry.Column> columnList = rowData.getAfterColumnsList();
                StringBuffer sql = new StringBuffer("INSERT INTO " + entry.getHeader().getTableName() + " (");
                for (int i = 0; i < columnList.size(); i++) {
                    sql.append(columnList.get(i).getName());
                    if (i != columnList.size() - 1) {
                        sql.append(", ");
                    }
                }
                sql.append(") VALUES (");
                for (int i = 0; i < columnList.size(); i++) {
                    sql.append("'" + columnList.get(i).getValue() + "'");
                    if (i != columnList.size() - 1) {
                        sql.append(", ");
                    }
                }
                sql.append(")");
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存删除语句
     *
     * @param entry
     */
    private void saveDeleteSql(CanalEntry.Entry entry) {
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDataList) {
                List<CanalEntry.Column> columnList = rowData.getBeforeColumnsList();
                StringBuffer sql = new StringBuffer("DELETE FROM " + entry.getHeader().getTableName() + " WHERE ");
                for (CanalEntry.Column column : columnList) {
                    if (column.getIsKey()) {
                        // 暂时只支持单一主键
                        sql.append(column.getName() + " = " + column.getValue());
                        break;
                    }
                }
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 保存更新语句
     *
     * @param entry
     */
    private void saveUpdateSql(CanalEntry.Entry entry) {
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            List<CanalEntry.RowData> rowDataList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDataList) {
                List<CanalEntry.Column> newColumnList = rowData.getAfterColumnsList();
                StringBuffer sql = new StringBuffer("UPDATE " + entry.getHeader().getTableName() + " SET ");
                for (int i = 0; i < newColumnList.size(); i++) {
                    sql.append(newColumnList.get(i).getName() + " = '" + newColumnList.get(i).getValue() + "'");
                    if (i != newColumnList.size() - 1) {
                        sql.append(", ");
                    }
                }
                sql.append(" WHERE ");
                List<CanalEntry.Column> oldColumnList = rowData.getBeforeColumnsList();
                for (CanalEntry.Column column : oldColumnList) {
                    if (column.getIsKey()) {
                        // 暂时只支持单一主键
                        sql.append(column.getName() + " = " + column.getValue());
                        break;
                    }
                }
                SQL_QUEUE.add(sql.toString());
            }
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
        }
    }

    /**
     * 模拟入库
     *
     * @param sql
     */
    public void execute(String sql) {
        log.info("模拟执行 SQL：{}", sql);
        /*Connection con = null;
        try {
            if (null == sql) return;
            con = dataSource.getConnection();
            QueryRunner qr = new QueryRunner();
            int row = qr.execute(con, sql);
            System.out.println("update: " + row);
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtils.closeQuietly(con);
        }*/
    }
}
